package org.eminem.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.eminem.hadoop.mapper.ESInitMapper;
import org.eminem.hbase.HbaseClient;
import org.eminem.utils.PropertiesUtil;

/**
 * Call Es init job
 */
public class ESInitCall extends Configured implements Tool {
	
	@Override
	public int run(String[] args) throws Exception {
		
		Configuration conf = this.getConf() ;
		
        Job job = Job.getInstance(conf, "ESInit Call Job");
        job.setJarByClass(ESInitCall.class);
        Scan scan = new Scan();
        scan.setCaching(500); // 1 is the default in Scan, which will be bad for MapReduce jobs
        scan.setCacheBlocks(false); // don't set to true for MR jobs
        
        //scan.setStartRow("00:global:gid:100000877".getBytes()) ;
        //scan.setStopRow("00:global:gid:100009646".getBytes()) ;
        
        // 设置map过程
        String tableName = PropertiesUtil.getStringByKey("hbase.histable.name");
        TableMapReduceUtil.initTableMapperJob(
                tableName,      // input table
                scan,              // Scan instance to control CF and attribute selection
                ESInitMapper.class,   // mapper class
                NullWritable.class,   // mapper output key
                NullWritable.class,   // mapper output value
                job);
        
        job.setNumReduceTasks(0);
        
        TableMapReduceUtil.initTableReducerJob(
                tableName,      // output table
                null,             // reducer class
                job);
        
        return job.waitForCompletion(true) ? 0 : 1 ;
	}

    public static void main(String args[]) throws Exception {
        
        System.out.println("target index:" + PropertiesUtil.getStringByKey("es.document.index"));
        System.out.println("target type:" + PropertiesUtil.getStringByKey("es.document.type"));
        System.out.println("target qxtable:" + PropertiesUtil.getStringByKey("hbase.qxtable.name"));
        System.out.println("es.cluster.name:" + PropertiesUtil.getStringByKey("es.cluster.name"));
        System.out.println("es.cluster.port:" + PropertiesUtil.getStringByKey("es.cluster.port"));
        System.out.println("es.cluster.host:" + PropertiesUtil.getStringByKey("es.cluster.host"));
        System.out.println("hbase.zookeeper.quorum:" + PropertiesUtil.getStringByKey("hbase.zookeeper.quorum"));
        System.out.println("hbase.zookeeper.property.clientPort:"+ PropertiesUtil.getStringByKey("hbase.zookeeper.property.clientPort"));
        
        Configuration conf = HbaseClient.getConfiguration(PropertiesUtil.getStringByKey("hbase.zookeeper.quorum"), PropertiesUtil.getStringByKey("hbase.zookeeper.property.clientPort"));
        String index = PropertiesUtil.getStringByKey("es.document.index") ;
		conf.set("zookeeper.znode.parent", "/hbase-unsecure");
        conf.set("index", index);
        conf.set("type", PropertiesUtil.getStringByKey("es.document.type"));
        conf.set("es.cluster.name", PropertiesUtil.getStringByKey("es.cluster.name"));
        conf.set("es.cluster.port", PropertiesUtil.getStringByKey("es.cluster.port"));
        conf.set("es.cluster.host", PropertiesUtil.getStringByKey("es.cluster.host"));
        conf.set("hbase.qxtable.name", PropertiesUtil.getStringByKey("hbase.qxtable.name"));
        
        int status = ToolRunner.run(conf,new ESInitCall(),args) ;
		System.exit(status);
    }

}
